import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkKafka {
  /*spark streaming共现 kafka 的消费者
 构建 sparkconf 本地运行，运行应用程序名称1)
  2)---》 streamingcontext ，加载配构建 sparkstreaming3)kafka 配置 broker ，key value ，group id，消费模式
  spark 链按kafka 订，topic，streamingcontext4)
  循环的形式 打印!处理5)
  开启ssc ，监控 kafka 数据6)
  */
  // 1)构建 sparkconf本地运行，运行应用程序名称
  def main(args: Array[String]): Unit ={
    val conf = new SparkConf().setMaster("local[*]").setAppName("helloSparkKafka")
    //StreamingContext 需要导入依赖
    // spark streaming 可以进行流式处理，微批次处理，间隔2秒
    val ssc = new StreamingContext(conf,Seconds(2))


    // spark 输出红色 info信息--> error
ssc.sparkContext.setLogLevel("error")

  val kfkaParams = Map[String,Object](
    "bootstrap.servers"->"192.168.109.100:9092",
    "key.deserializer"->classOf[StringDeserializer],
    "value.deserializer"->classOf[StringDeserializer],
    "group.id"->"niit",
    "enable.auto.commit"->(false:java.lang.Boolean)
  )

    //4)spark 链接kafka 订阅,topic，streamingcontext
    // topic name
  val topicName = Array("15homework")
  val streamRdd  =KafkaUtils.createDirectStream[String,String](
    ssc,
    PreferConsistent,
    Subscribe[String,String](topicName,kfkaParams)
  )
  //
    //返回kafka 返回的streamRdd(一段有时间间隔的RDD)
    streamRdd.foreachRDD(
    x=> {
      if (!x.isEmpty()) { // 判断是否为空，! 相反
        val line = x.map(_.value()) // 匿名函数
        line.foreach(println) //打印

      }
    }
    )

  ssc.start()
  ssc.awaitTermination()//监控kafka数据



}

}
